package com.blackto.lib.eventbus.rabbitmq;

import cn.hutool.json.JSONUtil;
import com.blackto.lib.eventbus.abstraction.IEventBus;
import com.blackto.lib.eventbus.abstraction.IIntegrationEventHandler;
import com.blackto.lib.eventbus.event.IntegrationEvent;
import com.blackto.lib.eventbus.IEventBusSubscriptionsManager;
import com.blackto.lib.eventbus.SubscriptionInfo;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ApplicationContext;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public class RabbitMQEventBus implements IEventBus {
    private final String EXCHANGE_NAME = "RabbitMQEventBus";

    private final String queueName;

    private final IEventBusSubscriptionsManager subscriptionsManager;

    private final RabbitTemplate rabbitTemplate;

    private final AmqpAdmin amqpAdmin;

    private final ApplicationContext context;

    public RabbitMQEventBus(String queueName,
                            IEventBusSubscriptionsManager subscriptionsManager,
                            RabbitTemplate rabbitTemplate,
                            AmqpAdmin amqpAdmin,
                            ApplicationContext context) {
        this.queueName = queueName;
        this.subscriptionsManager = subscriptionsManager;
        this.rabbitTemplate = rabbitTemplate;
        this.amqpAdmin = amqpAdmin;
        this.context = context;
    }

    @RabbitListener(queues = "${eventbus.queue.name}", ackMode = "MANUAL")
    private void consumerOnReceived(Message message, Channel channel) {
        // 获取事件名，并将消息解码
        String eventName = message.getMessageProperties().getHeader("routingKey");
        String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);

        try {
            if (messageBody.toLowerCase().contains("throw-fake-exception")) {
                throw new RuntimeException(
                        String.format("Fake exception requested: \"{%s}\"", messageBody));
            }

            processEvent(eventName, messageBody);

            // 确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception ex) {
            log.warn("----- ERROR Processing message \"{}\"", message, ex);
            // 发生异常时，可以选择拒绝消息或重新入队等操作
            try {
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void processEvent(String eventName, String message) {
        log.info("Processing RabbitMQ event: {}", eventName);

        if (!subscriptionsManager.hasSubscriptionsForEvent(eventName)) {
            log.warn("No subscription for RabbitMQ event: {}", eventName);
            return;
        }

        List<SubscriptionInfo> subscriptions = subscriptionsManager.getHandlersForEvent(eventName);
        for(SubscriptionInfo subscription : subscriptions){
            // 获得handler的类型，并通过spring容器获取其实例
            Class<?> handlerType = subscription.getHandlerType();
            Object handler = context.getBean(handlerType);
            if(handler == null) {
                continue;
            }
            // 获取handler的具体类型
            Class<?> concreteHandlerType = handler.getClass();

            // 获得event的具体类型和实例
            Class<? extends IntegrationEvent> eventType = subscriptionsManager.getEventTypeByName(eventName);
            IntegrationEvent event = JSONUtil.toBean(message, eventType);

            // 调用handle方法, 注意，这里不要强制类型转化后调，需要直获取实例对象后调，因为可能涉及到代理对象
            Method handleMethod = null;
            try {
                handleMethod = concreteHandlerType.getMethod("handle", eventType);
            } catch (NoSuchMethodException e) {
                log.warn("Error Handler Type: {}", handlerType.getSimpleName(), e);
                continue;
            }
            try {
                handleMethod.invoke(handler, event);
            } catch (IllegalAccessException | InvocationTargetException e) {
                log.warn("Error Handler Type: {}", handlerType.getSimpleName(), e);
            }
        }
    }




    @Override
    public void publish(IntegrationEvent event) {
        // 获取事件名
        String eventName = event.getClass().getSimpleName();

        log.info("RabbitMQ to publish event: {} ({})", event.getEventId().toString(), eventName);

        // 定义交换机
        DirectExchange exchange = new DirectExchange(EXCHANGE_NAME);
        amqpAdmin.declareExchange(exchange);

        log.info("Publishing event to RabbitMQ: {}", event.getEventId().toString());

        // 形成消息体
        byte[] body = JSONUtil.toJsonStr(event).getBytes(StandardCharsets.UTF_8);
        // 定义消息头并置入事件名
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setHeader("routingKey", eventName);

        // 创建消息
        Message message = new Message(body, messageProperties);

        // 发布消息到指定的交换机和路由键
        rabbitTemplate.send(EXCHANGE_NAME, eventName, message);
    }

    @Override
    public <TEvent extends IntegrationEvent, THandler extends IIntegrationEventHandler<TEvent>>
    void subscribe(Class<TEvent> TEventType, Class<THandler> THandlerType) {
        // 获取事件名
        String eventName = subscriptionsManager.getEventName(TEventType);

        // 检查是否已有订阅
        // 使用者微服务只关心自己订阅了哪些事件
        // 这里的实现类基于内存，即，微服务在自己的内存内维护自己的订阅信息
        boolean containsKey =
                subscriptionsManager.hasSubscriptionsForEvent(eventName);
        if (!containsKey) {
            // 没有订阅，定义交换机和队列，并进行绑定
            DirectExchange exchange = new DirectExchange(EXCHANGE_NAME);
            amqpAdmin.declareExchange(exchange);
            Queue queue = new Queue(queueName);
            amqpAdmin.declareQueue(queue);
            Binding binding = BindingBuilder
                    .bind(queue)
                    .to(exchange)
                    .with(eventName);
            amqpAdmin.declareBinding(binding);
        }

        log.info("Subscribing to event {} with {}", eventName, THandlerType.getSimpleName());

        subscriptionsManager
                .addSubscription(TEventType, THandlerType);
    }

    @Override
    public <TEvent extends IntegrationEvent, THandler extends IIntegrationEventHandler<TEvent>>
    void unsubscribe(Class<TEvent> TEventType, Class<THandler> THandlerType) {
        String eventName = subscriptionsManager.getEventName(TEventType);

        log.info("Unsubscribing from event {}", eventName);

        subscriptionsManager.removeSubscription(TEventType, THandlerType);

        amqpAdmin.removeBinding(BindingBuilder
                .bind(new Queue(queueName))
                .to(new DirectExchange(EXCHANGE_NAME))
                .with(eventName));
    }
}
